package com.shujia.mr.mapjoin;

import org.apache.commons.io.input.ReaderInputStream;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.PathExistsException;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.URI;
import java.util.HashMap;

public class MapJoinMapper extends Mapper<LongWritable, Text,NullWritable,Text> {
    HashMap<String,String> scoreMap = new HashMap<String,String>();

    @Override
    protected void setup(Mapper<LongWritable, Text, NullWritable, Text>.Context context) throws IOException, InterruptedException {
        // setup函数在每个MapTask中执行一次
        System.out.println("setup函数执行了一次");

        // 可以通过getCacheFiles获取到Driver端定义的CacheFile
        URI[] cacheFiles = context.getCacheFiles();

        // 在做数据处理过程中，要使用 防御式编程模式
        if(cacheFiles.length >= 1) {
            URI cacheFile = cacheFiles[0];


            // 通过HDFS操作对象,读取对应的数据
            Configuration configuration = context.getConfiguration();
            FileSystem fileSystem = FileSystem.get(configuration);
            FSDataInputStream open = fileSystem.open(new Path(cacheFile.getPath()));


            BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(open));
            String line = "";
            while ((line =bufferedReader.readLine())!=null){
                String[] split = line.split("\t");
                if (split.length >= 2){
                    String id = split[0];
                    String score = split[1];
                    scoreMap.put(id,score);
                }
            }

            // 关闭
            bufferedReader.close();
            open.close();


        }else {
            // 对于存在的问题可以直接抛出对应的异常
            throw new PathExistsException("您要获取的路径不存在...");
//            System.out.println("您要获取的路径不存在...");
        }

    }

    /*
        NullWritable 是Hadoop中的空类型
     */
    @Override
    protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, NullWritable, Text>.Context context) throws IOException, InterruptedException {
//        System.out.println("map函数执行了..");
        /*
            现在去实验 MapJoin 替代 ReduceJoin
                问题：
                    map函数是每一行数据执行一次，但是如果要关联两个文件中的数据，那么需要同时读取两个文件中的数据
                        ，之后再对关联条件进行做匹配 （ID = ID）,如何实现，单靠map函数肯定不行

                Mapper中除了map函数以外，还有setup函数，去查看setUp函数作用
                    setup函数在每个MapTask中执行一次
                    这时，可以通过IO流的方式将一个文件中的所有数据读取出来放在某个集合中保存，之后再去
                        map函数中读取另一个文件中的所有行数据，再对行做ID匹配，匹配到则关联，写出数据

                    1.成绩数据通过setUp函数加载到HashMap中
                    2.假设成绩数据是存放在HDFS中，那么就需要通过FileSystem形式获取到score数据
                    3.Score文件的路径
         */

        String lineStu = value.toString();
        String[] splits = lineStu.split(",");

        if (splits.length >= 5){
            String id = splits[0];
            // 如果使用get方法，key不存在，那么会报错
//            String s = scoreMap.get(id);
            String score = scoreMap.getOrDefault(id, null);
            context.write(NullWritable.get(),new Text(lineStu + ","+ score));
        }


    }
}
